package com.lvmama.rhino.analyze.client

import com.lvmama.rhino.common.entity.JobConfig
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.logging.LazyLogging
import com.lvmama.rhino.common.utils.methodset.CalculateFunction
import com.lvmama.rhino.common.utils.methodset.CalculateFunction._
import com.lvmama.rhino.common.utils.spark.SparkApplication
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions._
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.{LongType, StringType}


/**
  * Created by yuanxiaofeng on 2016/11/18.
  */
class PreProcessWireless(config: JobConfig, path: String, endPath: String) extends SparkApplication with LazyLogging{
  override var appName: String = "PreProcessWirelessStatJob"
  override var sparkConfig: Map[String, String] = config.spark

  def execute() = {
    sparkConfig += ("spark.app.name" -> appName)
    sparkConfig += ("spark.broadcast.blockSize" -> "200m")
    sparkConfig += ("spark.master" -> "local[3]")

    withSparkContext { sc =>
      val hiveCtx = new HiveContext(sc)
      import hiveCtx.implicits._
      val log = sc.textFile(path)
        .map(l => Utils.initWireless(l))
        .toDF()
        .transform(CalculateFunction.fieldVerify)

      import com.lvmama.rhino.common.utils.JDBCUtil.Conversions._
      val predicates = Array("CodeTypeLevel = 'D2'")
      val deviceCode = hiveCtx
        .loadFromMysql("statistic_continuity_code", predicates)

      val wap = deviceCode.filter(col("en_name") === "wap").select("code").collect()
      val PC = deviceCode.filter(col("en_name") === "PC").select("code").collect()

      val forward = log.filter(col("logType") === "forward" || col("logType") === "click")

      import com.lvmama.rhino.analyze.client.PreProcessWireless._

      val nativeDF = forward
        .filter(!$"platformCode".isin(wap.map(_.get(0)): _*))
        .filter(!$"platformCode".isin(PC.map(_.get(0)): _*))
        .coalesce(20)
        .withColumn("countByTimestamp", countByTimestamp)
        .withColumn("sessionId", traversalAssign(col("deviceToken"), col("countByTimestamp")).cast(StringType))
        .drop(col("countByTimestamp"))

      //页面访问事件
      val wapDF = forward
        .filter($"platformCode".isin(wap.map(_.get(0)): _*) || $"platformCode".isin(PC.map(_.get(0)): _*))
        .coalesce(5)

      val wapDF_bc = sc.broadcast(wapDF)

      val pageForward = nativeDF.coalesce(10)
      .union(wapDF_bc.value)

//      val processDF = pageForward.withColumn("timestamp", col("timestamp").cast(LongType))
      pageForward.write.mode(SaveMode.Append).parquet(endPath)
    }
  }
}

object PreProcessWireless {
  def apply(path: String, endPath: String) = {
    val config = JobConfig()
    new PreProcessWireless(config, config.param.get("wireless.source.path").get + path,
      config.param.get("wireless.process.path").get + endPath).execute()
  }

  val countByTimestamp = count("*").over(windowRange(col("deviceToken"), col("timestamp").cast(LongType)))
}
